/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.framework;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.singularity.io.BroadcastPipe;
import org.singularity.io.CloseablePipe;
import org.singularity.io.DME;
import org.singularity.io.Pump;
import org.singularity.mapred.Destination;
import org.singularity.mapred.Element;
import org.singularity.mapred.InputElement;
import org.singularity.mapred.OutputElement;
import org.singularity.mapred.ReduceElement;
import org.singularity.mapred.model.DefaultCascade;
import org.singularity.mapred.model.Flow;
import org.singularity.mapred.model.MapFlow;

/**
 * @author vjache
 *
 */
@SuppressWarnings("unchecked")
public class Pipeline implements Runnable {
	
	private static final Log LOG = LogFactory.getFactory().getInstance(Pipeline.class);

	private final DefaultCascade _cascade;
	private final HashMap<Element, CloseablePipe> _pipes = new HashMap<Element, CloseablePipe>(20);
	private final HashSet<Pump<DME>> _pumps = new HashSet<Pump<DME>>();
	
	private final AffinitiveExecutorService _exec = new AffinitiveExecutorService();

	/**
	 * 
	 */
	public Pipeline(DefaultCascade aCascade) {
		_cascade = aCascade;
	}
	
	private void build()
	{
		final ArrayList<Element> elems = new ArrayList<Element>();
		for(Element e:_cascade.topologyOrdered())
			elems.add(e);
		Collections.reverse(elems);
		
		for(Element e:elems)
		{
			buildElement(e);
		}
	}

	private void buildElement(Element e) {
		switch(e.getElementCode())
		{
		case Input:
		{
			buildElement((InputElement<?>)e);
			break;
		}
		case Output:
		{
			buildElement((OutputElement<?>)e);
			break;
		}
		case Reducer:
		{
			buildElement((ReduceElement<?,?>)e);
			break;
		}
		}
	}
	
	private void buildElement(InputElement aInp) {
		final Set<Flow> outflows = _cascade.outflows(aInp);
		for(Flow f:outflows)
		{
			final Destination destination = f.getDestination();
			switch(f.getFlowCode())
			{
			case MapFlow:
			{
				// 1. Create Pump
				// 2. Create Demultiplexor
				// 3. Take opposit element pipe
				// 4. Assemble
				final CloseablePipe redPipe = _pipes.get(destination);
				final MapperPipe mprPipe = new MapperPipe(aInp.getId(), redPipe, ((MapFlow)f).getMapFun());
				Pump<DME> inpPump = new Pump<DME>(aInp.getInput().iterator(), mprPipe, aInp+"->"+destination);
				_pumps.add(inpPump);
				break;
			}
			case OutputFlow:
			{
				final CloseablePipe redPipe = _pipes.get(destination);
				Pump<DME> inpPump = new Pump<DME>(aInp.getInput().iterator(), redPipe, aInp+"->"+destination);
				_pumps.add(inpPump);
				break;
			}
			}
		}
	}
	private void buildElement(OutputElement<?> aOut) {
		_pipes.put(aOut, aOut.getOutput().open());
	}
	private void buildElement(ReduceElement aRed) {
	//1. Analize outflows and create target pipe for this reducer
		final Set<Flow> outflows = _cascade.outflows(aRed);
		final HashSet<CloseablePipe> outPipes = new HashSet<CloseablePipe>(outflows.size());
		for(Flow f:outflows)
		{
			final CloseablePipe destPipe = _pipes.get(f.getDestination());
			switch(f.getFlowCode())
			{
				case MapFlow:{
					final MapperPipe mprPipe = new MapperPipe(aRed.getId(), destPipe, ((MapFlow)f).getMapFun());
					outPipes.add(mprPipe);
					break;
				}
				case OutputFlow:{
					outPipes.add(destPipe);
					break;
				}
			}
		}
		
		final CloseablePipe nextPipe;
		
		if(outPipes.size()>1)
			nextPipe = new BroadcastPipe(outPipes);
		else if(outPipes.size()==1)
			nextPipe = outPipes.iterator().next();
		else
			throw new RuntimeException(String.format("Reducer %s have no outflows.", aRed));
		
	//2. Discover mapper ids inflowing to this reducer and sort them by id(!).
		final Set<Flow> inflows = _cascade.inflows(aRed);
		final ArrayList mprIds = new ArrayList(inflows.size());
		for(Flow f:inflows)
		{
			mprIds.add(f.getSource().getId());
		}
		Collections.sort(mprIds);
	//
		
		final CloseablePipe redPipe = new ReducerPipe(nextPipe, mprIds, aRed.getFun(), createReduceCache(aRed,mprIds));
		
		ArrayList<CloseablePipe> redInPipes = createReducerInputPipes(aRed,
				redPipe);
		final HashedDemultiplexerPipe demulPipe = new HashedDemultiplexerPipe(redInPipes);
		_pipes.put(aRed, demulPipe);
	}

	private ArrayList<CloseablePipe> createReducerInputPipes(
			ReduceElement aRed, final CloseablePipe redPipe) {
		final int tNum = pumpsForReducer(aRed);
		ArrayList<CloseablePipe> redInPipes = new ArrayList<CloseablePipe>(tNum);
		for(int i = 0;i<tNum;i++)
		{
			final RunnablePipe b = new RunnablePipe(redPipe,_exec);
			redInPipes.add(b);
		}
		return redInPipes;
	}
//	private ArrayList<CloseablePipe> createReducerInputPipes(
//			ReduceElement aRed, final CloseablePipe redPipe) {
//		final int tNum = pumpsForReducer(aRed);
//		ArrayList<CloseablePipe> redInPipes = new ArrayList<CloseablePipe>(tNum);
//		for(int i = 0;i<tNum;i++)
//		{
//			final QueuedBlockingBridge b = new QueuedBlockingBridge(100);
//			redInPipes.add(b.open());
//			Pump<DME> redPump = new Pump<DME>(b.iterator(), redPipe, aRed);
//			_pumps.add(redPump);
//		}
//		return redInPipes;
//	}
	
	private int pumpsForReducer(ReduceElement aRed) {
		return 16;
	}

	private ReduceCache createReduceCache(ReduceElement aRed,Collection<Object> aMprIds)
	{
		return new DefaultReduceCache(aRed,aMprIds);
	}

	/* (non-Javadoc)
	 * @see java.lang.Runnable#run()
	 */
	@Override
	public void run() {
		build();
		final int pumpsNum = _pumps.size();
		final CountDownLatch latch = new CountDownLatch(pumpsNum);
		final ExecutorService tpool = Executors.newFixedThreadPool(pumpsNum,new ThreadFactory() {
			@Override
			public Thread newThread(final Runnable r) {
				return new Thread(new Runnable() {
					@Override
					public void run() {
						try{r.run();}
						finally
						{latch.countDown();}
					}
				}, r.toString());
			}
		});
		LOG.info(String.format("Start pumps %s ...",pumpsNum));
		int cnt=1;
		for(Pump<DME> pump:_pumps)
		{
			tpool.submit(pump);
			LOG.info(String.format("Pump %s (%s/%s) started.",pump,cnt,pumpsNum));
			cnt++;
		}
		LOG.info(String.format("Pumps started %s.",pumpsNum));
		try {
			latch.await();
			LOG.info("Pipeline finished.");
		} catch (InterruptedException e) {
			LOG.warn("Pipeline interrupted.");
		}
	}

}
